package com.jdrx.sw.sewage.service;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.jdrx.ocp.dao.UserDAO;
import com.jdrx.platform.commons.rest.beans.vo.ResponseVO;
import com.jdrx.platform.commons.rest.exception.BizException;
import com.jdrx.platform.commons.rest.factory.ResponseFactory2;
import com.jdrx.sw.sewage.beans.entity.*;
import com.jdrx.sw.sewage.beans.vo.AlarmSetVO;
import com.jdrx.sw.sewage.beans.vo.DeviceIotVO;
import com.jdrx.sw.sewage.beans.vo.DevicePointVO;
import com.jdrx.sw.sewage.beans.vo.StationVO;
import com.jdrx.sw.sewage.common.constant.Constant;
import com.jdrx.sw.sewage.common.enums.*;
import com.jdrx.sw.sewage.common.util.Asserts;
import com.jdrx.sw.sewage.common.util.LocalDateTimeUtils;
import com.jdrx.sw.sewage.common.util.NumberUtil;
import com.jdrx.sw.sewage.common.util.StringUtil;
import com.jdrx.sw.sewage.dao.*;
import com.jdrx.sw.sewage.service.base.BaseServiceImpl;
import com.jdrx.sw.sewage.service.common.RedisService;
import com.jdrx.sw.sewage.service.common.SequenceDefineServiceImpl;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Pattern;

import static com.jdrx.sw.sewage.common.constant.Constant.*;

@Service
public class AlarmDataHandleServiceImpl extends BaseServiceImpl {
    public static final Logger logger = LoggerFactory.getLogger(AlarmDataHandleServiceImpl.class);
    @Value("${redis.expire}")
    private long expire;
    @Autowired
    private RedisService redisService;
    @Autowired
    private AlarmHandleServiceImpl alarmHandleServiceImpl;
    @Autowired
    private SequenceDefineServiceImpl sequenceDefineServiceImpl;
    @Autowired
    private DeviceServiceImpl deviceServiceImpl;
    @Autowired
    private StationServiceImpl stationServiceImpl;
    @Autowired
    private DeviceTimeDataServiceImpl deviceTimeDataServiceImpl;
    @Autowired
    private AlarmDAO alarmDAO;
    @Autowired
    private AlarmSetDAO alarmSetDAO;
    @Autowired
    private AlarmCurveSnapDAO alarmCurveSnapDAO;
    @Autowired
    private GatherDevServiceImpl gatherDevServiceImpl;
    @Autowired
    private MessageSubscribeServiceImpl messageSubscribeServiceImpl;
    @Autowired
    private UserDAO userDAO;
    @Autowired
    private DevStorageServiceImpl devStorageServiceImpl;
    @Autowired
    private GatherDevPoiServiceImpl gatherDevPoiServiceImpl;
    @Autowired
    private MessageSubscribeDAO messageSubscribeDAO;
    @Autowired
    private PointCalculationRuleService pointCalculationRuleService;
    @Autowired
    private AiModelParamDAO aiModelParamDAO;

    /**
     * 厂站网络断开（在线状态）
     * <p>
     * redis key
     * aaa:status:devId
     * aaa:status:xxx
     * redis value
     * id:xxx
     * status:xxx
     * time:xxx
     *
     * @param list <p>
     *             [{
     *             id:222
     *             status:1,2
     *             time:'2019-10-10 10:10:10'
     *             }]
     *             </p>
     *             过期时间，永不过期   （1-在线 2-离线）
     * @throws BizException
     */
    @Transactional(rollbackFor = Exception.class)
    public void stationNoNetworkOutfcontactStatus(List<Map<String, Object>> list) throws BizException {
        Asserts.notEmpty(list, "接受到的设备状态消息");
        LocalDateTime now = LocalDateTime.now();

        // 组装解析后的值存入redis
//        List<Map<String, Map<byte[], byte[]>>> statusDataList = new ArrayList<>();
        try {
//            for (Map<String, Object> map : list) {
            for (int i = 0, len = list.size(); i < len; i++) {
                Map<String, Object> map = list.get(i);
                // 获取dataNode  tenantId为null则为一体机反之为云版本
                String tenantId = MapUtils.getString(map, "tenantId");

                String dataNode = null;
                if (StringUtils.isNotEmpty(tenantId)) {
                    Map<String, Object> mget = redisService.mget(REDIS_TENANT + tenantId, Object.class);
                    if (null == mget) {
                        logger.debug("未找到该{}租户id的dataNode,不处理告警", tenantId);
                        continue;
                    }
                    Map<String, Object> dataNodeMap = (Map<String, Object>) JSON.parse((String) mget.get("dataNode"));
                    dataNode = dataNodeMap.get(Constant.APPLICATION_NAME) + "";
                    if (StringUtils.isBlank(dataNode) || dataNode.equals("null")) {
                        logger.debug("该{}dataNode为空,不处理告警", dataNode);
                        continue;
                    }
                } /*else dataNode = MapUtils.getString(map, "tenantId");*/


                String status = MapUtils.getString(map, "status");

                if (Objects.isNull(status) || EDevOnLineStatus.NO_DATA.getKey() == Integer.valueOf(status).intValue()) {
                    logger.debug("{}站点状态为空,不处理告警", status);
                    continue;
                }

                //校验物联网网关和站点是否存在
                String netNum = MapUtils.getString(map, "id");
                StationVO stationVO = stationServiceImpl.selectStationByNetNum(netNum, getSQLPerfix(dataNode));
                if (Objects.isNull(stationVO)) {
                    logger.debug("{}的站点对象为空,不处理告警", stationVO);
                    continue;
                }
                Long stationId = stationVO.getId();
                String time = MapUtils.getString(map, "time");

                Map<String, Object> sensorTempMap = new HashMap<>();
                sensorTempMap.put("id", stationId.toString());
                sensorTempMap.put("status", status);
                sensorTempMap.put("time", time);

                String redisKey = DEVICE_STATUS_KEY + "_" + stationId + "_" + netNum;
                redisService.setMap(redisKey, sensorTempMap, null);
                // 离线才处理告警
                if (EDevOnLineStatus.OFF_LINE.getKey() == Integer.valueOf(status).intValue()) {
                    gatherDevPoiServiceImpl.updateByStationIdAndIotGateway(stationId, netNum);
                    // 有未处理的告警则不告警
                    AlarmPO alarmPO = alarmDAO.selectAlarmByDeviceIdAndAlarmStatus(stationId, EAlarmStatus.COMPLETED.getKey(), getSQLPerfix(dataNode), EAlarmDevType.STATION.getKey(), netNum);
                    if (Objects.nonNull(alarmPO)) continue;
                    addOutOfcontactAlarm(now, getSQLPerfix(dataNode), stationVO, STATION_OUTOFCONTACT_NETWORK_CH, netNum);
                }
                //在线才检测自动关闭告警
                if (devStatusSysTest(dataNode, status, netNum, stationId, redisKey)) continue;
            }
        } catch (Exception e) {
            logger.error("刷新物联网网关状态失败", e);
            throw new BizException("刷新物联网网关状态失败");
        }
    }

    /**
     * 系统关闭网络断开厂站
     *
     * @throws BizException
     */
    private boolean devStatusSysTest(String dataNode, String status, String netNum, Long stationId, String redisKey) throws Exception {
        if (EDevOnLineStatus.ON_LINE.getKey() == Integer.valueOf(status).intValue()) {
//            if (EDevOnLineStatus.ON_LINE.getKey() == (Objects.isNull(redisMap) ? EDevOnLineStatus.NO_DATA.getKey() : redisStatus) && 1 <= LocalDateTimeUtils.timeDiff(MapUtils.getString(redisMap, "time"), LocalDateTimeUtils.localDateTime2String(LocalDateTime.now()))) {
            // 没有未处理的告警则不告警
            AlarmPO alarmPO = alarmDAO.selectAlarmByDeviceIdAndAlarmStatus(stationId, EAlarmStatus.COMPLETED.getKey(), getSQLPerfix(dataNode), EAlarmDevType.STATION.getKey(), netNum);
            if (Objects.isNull(alarmPO)) return true;
            updateOutOfcontactAlarm(LocalDateTime.now(), getSQLPerfix(dataNode), alarmPO);
//            redisService.del(redisKey);
        }
        return false;
    }

    /**
     * 厂站欠费失联告警
     *
     * @throws BizException
     */
    @Transactional(rollbackFor = Exception.class)
    public void stationArrearsOutfcontact(String dataNode, StationVO station) throws BizException {
        if (!StringUtils.isEmpty(dataNode)) dataNode = getSQLPerfix(dataNode);

        AlarmPO alarmPO = alarmDAO.selectAlarmByDeviceIdAndAlarmStatus(station.getId(), EAlarmStatus.COMPLETED.getKey(), dataNode, EAlarmDevType.STATION.getKey(), null);
        // 有未处理的告警则不告警
        if (Objects.nonNull(alarmPO)) {
            logger.debug("{},已有未处理完成的失联告警，不处理告警", station.getName());
            return;
        }

        //做欠费失联告警处理
        addOutOfcontactAlarm(LocalDateTime.now(), dataNode, station, STATION_OUTOFCONTACT_ARREARS_CH, null);
    }

    @Transactional(rollbackFor = Exception.class)
    @Async("asyncServiceExecutor")
    public void sysCloseAlarmThread(String dataNode, String vaStr, DeviceIotVO deviceIotVO, String devPointValue) {
        sysCloseAlarm(dataNode, vaStr, deviceIotVO, devPointValue);
    }

    /**
     * 系统自动关闭-设备告警、水质达标告警
     *
     * @throws BizException
     */
    @Transactional(rollbackFor = Exception.class)
    public void sysCloseAlarm(String dataNode, String vaStr, DeviceIotVO deviceIotVO, String devPointValue) {
        BigDecimal val = new BigDecimal(NumberUtil.precision(vaStr, deviceIotVO.getPrecision()));
        String[] devPointValues = devPointValue.split("_");
        String devPointValue1 = devPointValues[2];

        String devPointValueNoNum = null;
        if (devPointValue1.contains(EQualityAlarmPar.NH3N.getKey())
                || devPointValue1.contains(EQualityAlarmPar.BOD5.getKey())) devPointValueNoNum = devPointValue1;
        else devPointValueNoNum = StringUtil.removeStrMatcher(devPointValue1);


        //处理设备告警 并且为故障和为正常值 才检测自动关闭告警
        if (EAlarmDict.DEVICE.getKey().equals(deviceIotVO.getAlarmTypeValue()) && ZG.equals(devPointValueNoNum) && FALSE == val.intValue())
            testData(dataNode, deviceIotVO, new BigDecimal(TRUE), null);

        if (EAlarmDict.QUALITY.getKey().equals(deviceIotVO.getAlarmTypeValue())
                && CSQ.equals(StringUtil.removeStrMatcher(devPointValues[0]))
                && 1 == val.compareTo(new BigDecimal(FALSE))
                && (EQualityAlarmPar.COD.getKey().equals(devPointValueNoNum)
                || EQualityAlarmPar.BOD5.getKey().equals(devPointValueNoNum)
                || EQualityAlarmPar.SS.getKey().equals(devPointValueNoNum)
                || EQualityAlarmPar.TP.getKey().equals(devPointValueNoNum)
                || EQualityAlarmPar.TN.getKey().equals(devPointValueNoNum)
                || EQualityAlarmPar.PH.getKey().equals(devPointValueNoNum)
                || EQualityAlarmPar.NH3N.getKey().equals(devPointValueNoNum))
        ) {
            AlarmSetVO alarmSetYeAlarmVO = alarmSetDAO.selectAlarmSetBySwitchAndEffectAndNameAndLevel(devPointValueNoNum, EAlarmLevel.ALARM_YELLOW.getKey(), dataNode);
            if (Objects.isNull(alarmSetYeAlarmVO)) return;

            if (EQualityAlarmPar.COD.getKey().equals(devPointValueNoNum) && -1 == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getCod())))
                testData(dataNode, deviceIotVO, new BigDecimal(alarmSetYeAlarmVO.getCod()), null);

            if (EQualityAlarmPar.BOD5.getKey().equals(devPointValueNoNum) && -1 == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getBod5())))
                testData(dataNode, deviceIotVO, new BigDecimal(alarmSetYeAlarmVO.getBod5()), null);

            if (EQualityAlarmPar.SS.getKey().equals(devPointValueNoNum) && -1 == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getSs())))
                testData(dataNode, deviceIotVO, new BigDecimal(alarmSetYeAlarmVO.getSs()), null);

            if (EQualityAlarmPar.TP.getKey().equals(devPointValueNoNum) && -1 == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getTp())))
                testData(dataNode, deviceIotVO, new BigDecimal(alarmSetYeAlarmVO.getTp()), null);

            if (EQualityAlarmPar.TN.getKey().equals(devPointValueNoNum) && -1 == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getTn())))
                testData(dataNode, deviceIotVO, new BigDecimal(alarmSetYeAlarmVO.getTn()), null);

            //ph单独处理
            if (EQualityAlarmPar.PH.getKey().equals(devPointValueNoNum)
                    && (1 == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getPhMin()))
                    && -1 == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getPhMax())))) {
                testData(dataNode, deviceIotVO, new BigDecimal(alarmSetYeAlarmVO.getPhMin()), new BigDecimal(alarmSetYeAlarmVO.getPhMax()));
            }

            //NH3N处理  北半球4-9 夏令时  10-3是冬令时
            if (EQualityAlarmPar.NH3N.getKey().equals(devPointValueNoNum)) {
                int seasonNumber = Calendar.getInstance().get(Calendar.MONTH);
                if ((seasonNumber >= 4 && seasonNumber <= 9) && -1 == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getNh3NS())))
                    testData(dataNode, deviceIotVO, new BigDecimal(alarmSetYeAlarmVO.getNh3NS()), null);

                if ((seasonNumber >= 10 || seasonNumber <= 3) && -1 == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getNh3NW())))
                    testData(dataNode, deviceIotVO, new BigDecimal(alarmSetYeAlarmVO.getNh3NW()), null);
            }
        }
    }

    private void testData(String dataNode, DeviceIotVO deviceIotVO, BigDecimal val, BigDecimal val2) {
        AlarmPO alarmPO = alarmDAO.selectAlarmByDeviceIdAndAlarmStatus(deviceIotVO.getDeviceId(), EAlarmStatus.COMPLETED.getKey(), dataNode, EAlarmDevType.DEV.getKey(), null);
        //检查是否有未处理的告警
        if (Objects.isNull(alarmPO)) {
            logger.info("{},没有未处理的告警，不处理自动关闭", deviceIotVO.getGatherDevPointId());
            return;
        }
        //当前值为正常值才检测
        Integer count = deviceTimeDataServiceImpl.sysTestCloseAlarm(deviceIotVO.getGatherDevPointId(),
                alarmPO.getStartTime(),
                deviceIotVO.getGatherDevPointValue(), val, val2, dataNode);
        //关闭告警
        if (0 == count) updateOutOfcontactAlarm(LocalDateTime.now(), dataNode, alarmPO);
    }

    /**
     * 系统自动关闭-厂站欠费失联告警
     *
     * @throws BizException
     */
    @Transactional(rollbackFor = Exception.class)
    public void sysCloseStationArrearsOutfcontact(String dataNode, StationVO stationVO) throws BizException {
        if (!StringUtils.isEmpty(dataNode)) dataNode = getSQLPerfix(dataNode);

        AlarmPO alarmPO = alarmDAO.selectAlarmByDeviceIdAndAlarmStatus(stationVO.getId(), EAlarmStatus.COMPLETED.getKey(), dataNode, EAlarmDevType.STATION.getKey(), null);
        // 没有有未处理的告警则不关闭告警
        if (Objects.isNull(alarmPO)) {
            logger.debug("{},没有未处理完成的失联告警，不关闭告警", stationVO.getName());
            return;
        }

        //关闭欠费失联告警处理
        updateOutOfcontactAlarm(LocalDateTime.now(), dataNode, alarmPO);
    }

    /**
     * 用户关闭-设备告警
     *
     * @throws BizException
     */
    @Transactional(rollbackFor = Exception.class)
    public ResponseVO userCloseDeviceAlarm(AlarmPO alarmPO, String userName) throws BizException {
        //拿到设备命名规范
        DevicePointVO devicePointVO = alarmDAO.findDevPointValueByAlarmIdAndDevTypeA(alarmPO.getId(), alarmPO.getDevType(), alarmPO.getGatherDevPoiId());

        if (null == devicePointVO) return ResponseFactory2.err("设备不存在，不能关闭告警");

        /*Integer count = deviceTimeDataServiceImpl.sysTestCloseAlarm(alarmPO.getDeviceId(),
                alarmPO.getStartTime(),
                devicePointVO.getDevPointValue(), new BigDecimal(FALSE), null, null);

        //关闭告警
        if (FALSE < count) Asserts.check(false, "设备近一个小时内仍有故障状态，不能关闭告警");*/
        Double parValue = devicePointVO.getParValue();
        if (null == parValue || FALSE == new BigDecimal(TRUE).compareTo(new BigDecimal(parValue)))
            return ResponseFactory2.err("设备仍有故障状态，不能关闭告警");

        userUpdateOutOfcontactAlarm(LocalDateTime.now(), alarmPO, userName);

        return ResponseFactory2.ok("关闭成功");
    }

    /**
     * 用户关闭-水质达标告警
     *
     * @throws BizException
     */
    @Transactional(rollbackFor = Exception.class)
    public ResponseVO userCloseQualityAlarm(AlarmPO alarmPO, String userName) throws BizException {
        //拿到设备命名规范
        DevicePointVO devicePointVO = alarmDAO.findDevPointValueByAlarmIdAndDevTypeA(alarmPO.getId(), alarmPO.getDevType(), alarmPO.getGatherDevPoiId());
        if (null == devicePointVO) return ResponseFactory2.err("设备不存在，不能关闭告警");

        //拿到设置标准值
        String[] devPointValues = devicePointVO.getDevPointValue().split("_");
        String devPointValue = devPointValues[devPointValues.length - 2];

        Double parValue = devicePointVO.getParValue();
        if (null == parValue) return ResponseFactory2.err(devPointValue + "值仍处于异常，不能关闭告警");

        if (devPointValue.contains(NH3N)) devPointValue = NH3N;
        if (devPointValue.contains(BOD5)) devPointValue = BOD5;

        if (!devPointValue.contains(NH3N) && !devPointValue.contains(BOD5))
            devPointValue = StringUtil.removeStrMatcher(devPointValue);

        AlarmSetVO alarmSetYeAlarmVO = alarmSetDAO.selectAlarmSetBySwitchAndEffectAndNameAndLevel(devPointValue, EAlarmLevel.ALARM_YELLOW.getKey(), null);

        if (Objects.isNull(alarmSetYeAlarmVO)) {
            userUpdateOutOfcontactAlarm(LocalDateTime.now(), alarmPO, userName);
            return ResponseFactory2.ok("关闭成功");
        }

        BigDecimal val = new BigDecimal(parValue);
        if (EQualityAlarmPar.COD.getKey().equals(devPointValue)
                && TRUE == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getCod())))
            return ResponseFactory2.err("COD值仍处于异常，不能关闭告警");

        if (EQualityAlarmPar.BOD5.getKey().equals(devPointValue)
                && TRUE == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getBod5())))
            return ResponseFactory2.err("BOD5值仍处于异常，不能关闭告警");

        if (EQualityAlarmPar.SS.getKey().equals(devPointValue)
                && TRUE == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getSs())))
            return ResponseFactory2.err("SS值仍处于异常，不能关闭告警");

        if (EQualityAlarmPar.TP.getKey().equals(devPointValue)
                && TRUE == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getTp())))
            return ResponseFactory2.err("TP值仍处于异常，不能关闭告警");

        if (EQualityAlarmPar.TN.getKey().equals(devPointValue)
                && TRUE == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getTn())))
            return ResponseFactory2.err("TN值仍处于异常，不能关闭告警");

        if (EQualityAlarmPar.PH.getKey().equals(devPointValue)
                && (TRUE == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getPhMax()))
                || MINUS_ONE == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getPhMin()))))
            return ResponseFactory2.err("PH值仍处于异常，不能关闭告警");

        if (EQualityAlarmPar.NH3N.getKey().equals(devPointValue)) {
            int seasonNumber = Calendar.getInstance().get(Calendar.MONTH) + 1;
            if (((seasonNumber >= 4 && seasonNumber <= 9)
                    && TRUE == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getNh3NS()))
            ) || ((seasonNumber >= 10 || seasonNumber <= 3)
                    && TRUE == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getNh3NW()))))
                return ResponseFactory2.err("NH3N值仍处于异常，不能关闭告警");
        }

        userUpdateOutOfcontactAlarm(LocalDateTime.now(), alarmPO, userName);

        return ResponseFactory2.ok();
    }

    /**
     * 用户关闭-水质达标告警
     *
     * @throws BizException
     */
    @Transactional(rollbackFor = Exception.class)
    public void userCloseQualityAlarm2(AlarmPO alarmPO, String userName) throws BizException {
        //拿到设备命名规范
        DevicePointVO devicePointVO = alarmDAO.findDevPointValueByAlarmIdAndDevTypeA(alarmPO.getId(), alarmPO.getDevType(), alarmPO.getGatherDevPoiId());
        Asserts.check(Objects.nonNull(devicePointVO), "设备不存在，不能关闭告警");

        //拿到设置标准值
        String[] devPointValues = devicePointVO.getDevPointValue().split("_");
        String devPointValue = StringUtil.removeStrMatcher(devPointValues[devPointValues.length - 2]);
        AlarmSetVO alarmSetYeAlarmVO = alarmSetDAO.selectAlarmSetBySwitchAndEffectAndNameAndLevel(devPointValue, EAlarmLevel.ALARM_YELLOW.getKey(), null);
        Integer count = null;
        if (EQualityAlarmPar.COD.getKey().equals(devPointValue))
            count = deviceTimeDataServiceImpl.sysTestCloseAlarm(alarmPO.getDeviceId(),
                    alarmPO.getStartTime(),
                    devicePointVO.getDevPointValue(), new BigDecimal(alarmSetYeAlarmVO.getCod()), null, null);

        if (EQualityAlarmPar.BOD5.getKey().equals(devPointValue))
            count = deviceTimeDataServiceImpl.sysTestCloseAlarm(alarmPO.getDeviceId(),
                    alarmPO.getStartTime(),
                    devicePointVO.getDevPointValue(), new BigDecimal(alarmSetYeAlarmVO.getBod5()), null, null);

        if (EQualityAlarmPar.SS.getKey().equals(devPointValue))
            count = deviceTimeDataServiceImpl.sysTestCloseAlarm(alarmPO.getDeviceId(),
                    alarmPO.getStartTime(),
                    devicePointVO.getDevPointValue(), new BigDecimal(alarmSetYeAlarmVO.getSs()), null, null);

        if (EQualityAlarmPar.TP.getKey().equals(devPointValue))
            count = deviceTimeDataServiceImpl.sysTestCloseAlarm(alarmPO.getDeviceId(),
                    alarmPO.getStartTime(),
                    devicePointVO.getDevPointValue(), new BigDecimal(alarmSetYeAlarmVO.getTp()), null, null);

        if (EQualityAlarmPar.TN.getKey().equals(devPointValue))
            count = deviceTimeDataServiceImpl.sysTestCloseAlarm(alarmPO.getDeviceId(),
                    alarmPO.getStartTime(),
                    devicePointVO.getDevPointValue(), new BigDecimal(alarmSetYeAlarmVO.getTn()), null, null);

        //ph单独处理
        if (EQualityAlarmPar.PH.getKey().equals(devPointValue))
            count = deviceTimeDataServiceImpl.sysTestCloseAlarm(alarmPO.getDeviceId(),
                    alarmPO.getStartTime(),
                    devicePointVO.getDevPointValue(), new BigDecimal(alarmSetYeAlarmVO.getPhMin()), new BigDecimal(alarmSetYeAlarmVO.getPhMax()), null);

        //NH3N处理
        if (EQualityAlarmPar.NH3N.getKey().equals(devPointValue)) {
            int seasonNumber = Calendar.getInstance().get(Calendar.MONTH);
            if ((seasonNumber >= 4 && seasonNumber <= 9))
                count = deviceTimeDataServiceImpl.sysTestCloseAlarm(alarmPO.getDeviceId(),
                        alarmPO.getStartTime(),
                        devicePointVO.getDevPointValue(), new BigDecimal(alarmSetYeAlarmVO.getNh3NS()), null, null);
            if (seasonNumber >= 10 || seasonNumber <= 3)
                count = deviceTimeDataServiceImpl.sysTestCloseAlarm(alarmPO.getDeviceId(),
                        alarmPO.getStartTime(),
                        devicePointVO.getDevPointValue(), new BigDecimal(alarmSetYeAlarmVO.getNh3NW()), null, null);
        }

        //关闭告警
        if (FALSE < count) Asserts.check(false, "设备近一个小时内数据仍不标准，不能关闭告警");
        userUpdateOutOfcontactAlarm(LocalDateTime.now(), alarmPO, userName);
    }

    /**
     * 用户关闭网络断开厂站
     *
     * @throws BizException
     */
    public ResponseVO devStatusUserTest(AlarmPO alarmPO, String userName, String dataNode) throws BizException {
        StationPO stationVO = stationServiceImpl.findById(alarmPO.getStationId());
        if (null == stationVO) return ResponseFactory2.err("厂站不存在，不能关闭告警");
        String redisKey = DEVICE_STATUS_KEY + "_" + alarmPO.getStationId() + "_" + alarmPO.getIotGatewayNum();

        Map<String, Object> redisMap = redisService.getMap(redisKey);
        if (redisMap == null || EDevOnLineStatus.OFF_LINE.getKey() == MapUtils.getInteger(redisMap, "status"))
            return ResponseFactory2.err("厂站仍处于网络断开，不能关闭告警");

        userUpdateOutOfcontactAlarm(LocalDateTime.now(), alarmPO, userName);
        redisService.del(redisKey);
        return ResponseFactory2.ok("关闭成功");
    }

    /**
     * 用户关闭-厂站欠费失联告警
     *
     * @throws BizException
     */
    @Transactional(rollbackFor = Exception.class)
    public ResponseVO userCloseStationArrearsOutfcontact(AlarmPO alarmPO, String userName) throws BizException {
        StationVO stationVO = stationServiceImpl.selectStationAlarmTypeById(alarmPO.getStationId(), null, StationUseStatus.ARREARS.getKey());
        if (null == stationVO) return ResponseFactory2.err("厂站仍然处于欠费状态，不能关闭告警");
        //关闭欠费失联告警处理
        userUpdateOutOfcontactAlarm(LocalDateTime.now(), alarmPO, userName);
        return ResponseFactory2.ok("关闭成功");
    }

    /**
     * 生成告警记录
     */
    @Transactional(rollbackFor = Exception.class)
    protected void generateAlarm(String dataNode, LocalDateTime now, String vaStr, DeviceIotVO deviceIotVO, String devPointValue) throws Exception {
        String alarm = Optional.ofNullable(deviceIotVO.getAlarmTypeValue()).orElse(EMPTY);
        BigDecimal val = new BigDecimal(NumberUtil.precision(vaStr, deviceIotVO.getPrecision(), ZEROS));

        //根据告警类型处理告警信息（目前除了水质告警分为黄色告警和红色告警，其他的都是红色告警）
        if (EAlarmDict.DEVICE.getKey().equals(alarm) && TRUE == val.intValue()) {
            //解析规范编号是设备故障告警才处理
            if (!ZG.equals(StringUtil.removeStrMatcher(devPointValue.split("_")[2]))) {
                logger.info("{},不是故障，不处理告警", deviceIotVO.getDeviceName());
                return;
            }

            AlarmPO alarmPO = alarmDAO.selectAlarmByDeviceIdAndAlarmStatus(deviceIotVO.getDeviceId(), EAlarmStatus.COMPLETED.getKey(), dataNode, EAlarmDevType.DEV.getKey(), null);
            //如果有，没有处理完的告警
            if (Objects.nonNull(alarmPO)) {
                logger.info("{},已有未处理完成的告警，不处理告警", deviceIotVO.getDeviceName());
                return;
            }
            List<Double> dv = deviceTimeDataServiceImpl.selectNearbyFiveByDevId(deviceIotVO.getGatherDevPointId(), deviceIotVO.getGatherDevPointValue(), dataNode);
            if (dv.size() < 4) return;
            for (int i = 0; i < dv.size(); i++) {
                if (dv.get(i) == 0) return;
            }
            deviceAlarmAddAlarmInfo(now, deviceIotVO, dataNode);
        }

        //水质告警，则和标准值判断，处理告警曲线快照
        String[] devPointValues = devPointValue.split("_");
        if (CSQ.equals(StringUtil.removeStrMatcher(devPointValues[0])) && EAlarmDict.QUALITY.getKey().equals(alarm) && 1 == val.compareTo(new BigDecimal(FALSE)))
            qualityAlarmAddAlarminfo(now, deviceIotVO, val, devPointValues[2], dataNode);
    }

    private void updateOutOfcontactAlarm(LocalDateTime now, String dataNode, AlarmPO alarmPO) {
        alarmPO.setEndTime(now);
        alarmPO.setDataNode(dataNode);
        alarmPO.setAlarmStatus(EAlarmStatus.COMPLETED.getKey());
        alarmDAO.updateAlarmInfoAndLevelAndEndTimeById(alarmPO);
        //处理系统消息
        handleAlarmSysNews(Constant.SYS_CLOSE_STATION_OUTOFCONTACT_ARREARS_CH, alarmPO.getId(), EAlarmHandleType.SYSTEM_HANDLE.getKey(), dataNode);
    }

    private void userUpdateOutOfcontactAlarm(LocalDateTime now, AlarmPO alarmPO, String userName) {
        alarmPO.setEndTime(now);
        alarmPO.setAlarmStatus(EAlarmStatus.COMPLETED.getKey());
        alarmDAO.updateAlarmInfoAndLevelAndEndTimeById(alarmPO);
        //处理系统消息
        alarmHandleServiceImpl.handleAlarmSysNews(Constant.USER_CLOSE_STATION_OUTOFCONTACT_ARREARS_CH,
                alarmPO.getId(), EAlarmHandleType.DUTY_USER_HANDLE.getKey(), null, userName, null);
    }

    private void addOutOfcontactAlarm(LocalDateTime now, String dataNode, StationVO stationVO, String alarmInfo, String netNum) throws BizException {
        AlarmPO alarmPO = new AlarmPO();
        alarmPO.setStationId(stationVO.getId());
        alarmPO.setStationName(stationVO.getName());
        alarmPO.setAlarmStatus(EAlarmStatus.UNREAD.getKey());
        alarmPO.setAlarmType(stationVO.getAlarmTypeId());
        alarmPO.setAlarmTypeName(stationVO.getAlarmTypeName());
        if (StringUtils.isEmpty(netNum))
            alarmPO.setAlarmInfo(String.format(alarmInfo, stationVO.getName()));
        else alarmPO.setAlarmInfo(String.format(alarmInfo, stationVO.getName(), netNum));
        alarmPO.setAlarmLevel(EAlarmLevel.ALARM_RED.getKey());
        alarmPO.setStartTime(now);
        alarmPO.setDutyUserNames(stationVO.getDutyUserNames());
        alarmPO.setDeviceId(stationVO.getId());
        alarmPO.setAlarmNum(String.format("SL" + LocalDateTimeUtils.localDateToString(new Date()) + "%03d", sequenceDefineServiceImpl.increment(Constant.ALARM_NUM_SN_KEY, dataNode)));
        alarmPO.setDataNode(dataNode);
        alarmPO.setDevType(EAlarmDevType.STATION.getKey());
        alarmPO.setIotGatewayNum(netNum);

        Long aId = alarmDAO.selsectNextAlarmId(dataNode);
        alarmPO.setId(aId);
        alarmDAO.insert(alarmPO);
        //处理系统消息
        handleAlarmSysNews(Constant.RED_ALARM_HANDLE_SYS_IDEA, alarmPO.getId(), EAlarmHandleType.SYSTEM_HANDLE.getKey(), dataNode);

        //push message
        try {
            List<Long> userIds = messageSubscribeDAO.selectMessageByStationIdAndType(stationVO.getId(), MessageSubscribeType.ALARM_EVENT_PUSH.getKey(), dataNode);
            if (null != userIds && userIds.size() > 0) {
                List<String> openId = userDAO.findOpenId(userIds, dataNode);
                openId.forEach(item -> {
                    try {
                        if (StringUtils.isNotEmpty(item))
                            messageSubscribeServiceImpl.pushAlarm(item,
                                    aId,
                                    alarmPO.getAlarmInfo(),
                                    "告警类型告警通知",
                                    "请尽快处理，点击查看详情",
                                    MessageSubscribeType.ALARM_EVENT_PUSH.getKey());
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
            }
        } catch (Exception e) {
            logger.info("告警推送失败！", e);
        }
    }

    private void deviceAlarmAddAlarmInfo(LocalDateTime now, DeviceIotVO deviceIotVO, String dataNode) throws Exception {
        AlarmPO alarmPO = new AlarmPO();
        alarmPO.setStationId(deviceIotVO.getStationId());
        alarmPO.setStationName(deviceIotVO.getStationName());
        alarmPO.setAlarmStatus(EAlarmStatus.UNREAD.getKey());
        alarmPO.setAlarmType(deviceIotVO.getAlarmTypeId());
        alarmPO.setAlarmTypeName(deviceIotVO.getAlarmTypeName());
        alarmPO.setAlarmInfo(String.format(GZ_CH, deviceIotVO.getDeviceName()));
        alarmPO.setAlarmLevel(EAlarmLevel.ALARM_RED.getKey());
        alarmPO.setStartTime(now);
        alarmPO.setDutyUserNames(deviceIotVO.getDutyUserNames());
        alarmPO.setDeviceId(deviceIotVO.getDeviceId());
        alarmPO.setAlarmNum(String.format("SB" + LocalDateTimeUtils.localDateToString(new Date()) + "%03d", sequenceDefineServiceImpl.increment(Constant.ALARM_NUM_SN_KEY, dataNode)));
        alarmPO.setDataNode(dataNode);
        alarmPO.setDevType(EAlarmDevType.DEV.getKey());
        alarmPO.setGatherDevPoiId(deviceIotVO.getGatherDevPointId());

        Long aId = alarmDAO.selsectNextAlarmId(dataNode);
        alarmPO.setId(aId);
        alarmDAO.insert(alarmPO);
        //处理系统消息
        handleAlarmSysNews(Constant.RED_ALARM_HANDLE_SYS_IDEA, alarmPO.getId(), EAlarmHandleType.SYSTEM_HANDLE.getKey(), dataNode);

        //push message
        try {
            List<Long> userIds = messageSubscribeDAO.selectMessageByStationIdAndType(deviceIotVO.getStationId(), MessageSubscribeType.ALARM_EVENT_PUSH.getKey(), dataNode);
            if (null != userIds && userIds.size() > 0) {
                List<String> openId = userDAO.findOpenId(userIds, dataNode);
                //        List<String> openId = userDAO.findOpenId(Arrays.asList((Long[]) ConvertUtils.convert(deviceIotVO.getOpeLeaderId().split(","), Long.class)), null);
                openId.forEach(item -> {
                    try {
                        if (StringUtils.isNotEmpty(item))
                            messageSubscribeServiceImpl.pushAlarm(item,
                                    aId,
                                    alarmPO.getAlarmInfo(),
                                    "告警类型告警通知",
                                    "请尽快处理，点击查看详情",
                                    MessageSubscribeType.ALARM_EVENT_PUSH.getKey());
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
            }
        } catch (Exception e) {
            logger.error("告警推送失败", e);
        }
    }

    private void qualityAlarmAddAlarmCurveSnap(LocalDateTime now, BigDecimal alarmValue, Long alarmId, String dataNode) throws BizException {
        AlarmCurveSnapPO alarmCurveSnapPO = new AlarmCurveSnapPO();
        alarmCurveSnapPO.setAlarmId(alarmId);
        alarmCurveSnapPO.setAlarmValue(alarmValue.doubleValue());
        alarmCurveSnapPO.setUpdateTime(now);
        alarmCurveSnapPO.setDataNode(dataNode);
        alarmCurveSnapDAO.insert(alarmCurveSnapPO);
    }

    private void qualityAlarmAddAlarminfo(LocalDateTime now, DeviceIotVO deviceIotVO, BigDecimal val, String devPointValue, String dataNode) throws BizException {
        String devPointValueNoNum = null;
        if (devPointValue.contains(EQualityAlarmPar.NH3N.getKey())
                || devPointValue.contains(EQualityAlarmPar.BOD5.getKey())) devPointValueNoNum = devPointValue;
        else devPointValueNoNum = StringUtil.removeStrMatcher(devPointValue);

        if (EAlarmDict.QUALITY.getKey().equals(deviceIotVO.getAlarmTypeValue())
                && 1 == val.compareTo(new BigDecimal(FALSE))
                && (EQualityAlarmPar.COD.getKey().equals(devPointValueNoNum)
                || EQualityAlarmPar.BOD5.getKey().equals(devPointValueNoNum)
                || EQualityAlarmPar.SS.getKey().equals(devPointValueNoNum)
                || EQualityAlarmPar.TP.getKey().equals(devPointValueNoNum)
                || EQualityAlarmPar.TN.getKey().equals(devPointValueNoNum)
                || EQualityAlarmPar.PH.getKey().equals(devPointValueNoNum)
                || EQualityAlarmPar.NH3N.getKey().equals(devPointValueNoNum))
        ) {
            AlarmSetVO alarmSetRedAlarmVO = alarmSetDAO.selectAlarmSetBySwitchAndEffectAndNameAndLevel(devPointValueNoNum, EAlarmLevel.ALARM_RED.getKey(), dataNode);
            AlarmSetVO alarmSetYAlarmVO = alarmSetDAO.selectAlarmSetBySwitchAndEffectAndNameAndLevel(devPointValueNoNum, EAlarmLevel.ALARM_YELLOW.getKey(), dataNode);
            if (Objects.isNull(alarmSetRedAlarmVO)) return;

            List<Double> dv = deviceTimeDataServiceImpl.selectNearbyFiveByDevId(deviceIotVO.getGatherDevPointId(), deviceIotVO.getGatherDevPointValue(), dataNode);
            if (dv.size() < 4) return;

            int seasonNumber = Calendar.getInstance().get(Calendar.MONTH);
            if (EQualityAlarmPar.NH3N.getKey().equals(devPointValueNoNum)) {
                //北半球4-9 夏令时  10-3是冬令时
                Double nh3N = null;
                Double nh3NY = null;
                if (seasonNumber >= 4 && seasonNumber <= 9) {
                    nh3N = alarmSetRedAlarmVO.getNh3NS();
                    nh3NY = alarmSetYAlarmVO.getNh3NS();
                } else {
                    nh3N = alarmSetRedAlarmVO.getNh3NW();
                    nh3NY = alarmSetYAlarmVO.getNh3NW();
                }

                for (int i = 0; i < dv.size(); i++) if (-1 == dv.get(i).compareTo(nh3NY)) return;
                qualityAlarmNotPh(now, deviceIotVO, val, devPointValue, seasonNumber, nh3N, dataNode);
            }

            if (EQualityAlarmPar.COD.getKey().equals(devPointValueNoNum)) {
                Double cod = alarmSetRedAlarmVO.getCod();
                for (int i = 0; i < dv.size(); i++) {
                    if (-1 == dv.get(i).compareTo(alarmSetYAlarmVO.getCod())) return;
                }
                qualityAlarmNotPh(now, deviceIotVO, val, devPointValue, seasonNumber, cod, dataNode);
            }

            if (EQualityAlarmPar.BOD5.getKey().equals(devPointValueNoNum)) {
                Double bod5 = alarmSetRedAlarmVO.getBod5();
                for (int i = 0; i < dv.size(); i++) if (-1 == dv.get(i).compareTo(alarmSetYAlarmVO.getBod5())) return;
                qualityAlarmNotPh(now, deviceIotVO, val, devPointValue, seasonNumber, bod5, dataNode);
            }

            if (EQualityAlarmPar.SS.getKey().equals(devPointValueNoNum)) {
                Double ss = alarmSetRedAlarmVO.getSs();
                for (int i = 0; i < dv.size(); i++) if (-1 == dv.get(i).compareTo(alarmSetYAlarmVO.getSs())) return;
                qualityAlarmNotPh(now, deviceIotVO, val, devPointValue, seasonNumber, ss, dataNode);
            }

            if (EQualityAlarmPar.TP.getKey().equals(devPointValueNoNum)) {
                Double tp = alarmSetRedAlarmVO.getTp();
                for (int i = 0; i < dv.size(); i++) if (-1 == dv.get(i).compareTo(alarmSetYAlarmVO.getTp())) return;
                qualityAlarmNotPh(now, deviceIotVO, val, devPointValue, seasonNumber, tp, dataNode);
            }

            if (EQualityAlarmPar.TN.getKey().equals(devPointValueNoNum)) {
                Double tn = alarmSetRedAlarmVO.getTn();
                for (int i = 0; i < dv.size(); i++) if (-1 == dv.get(i).compareTo(alarmSetYAlarmVO.getTn())) return;
                qualityAlarmNotPh(now, deviceIotVO, val, devPointValue, seasonNumber, tn, dataNode);
            }

            //ph单独处理
            if (EQualityAlarmPar.PH.getKey().equals(devPointValueNoNum)) {
//                AlarmSetVO alarmSetYeeAlarmVO = alarmSetDAO.selectAlarmSetBySwitchAndEffectAndNameAndLevel(devPointValue, EAlarmLevel.ALARM_RED.getKey(), dataNode);
                for (int i = 0; i < dv.size(); i++) {
                    Double v = dv.get(i);
                    if (-1 == v.compareTo(alarmSetYAlarmVO.getPhMax()) && 1 == v.compareTo(alarmSetYAlarmVO.getPhMin()))
                        return;
                }
                qualityAlarmPh(now, deviceIotVO, val, devPointValue, dataNode, alarmSetRedAlarmVO);
            }
        }
    }

    private void qualityAlarmPh(LocalDateTime now, DeviceIotVO deviceIotVO, BigDecimal val, String devPointValue, String dataNode, AlarmSetVO alarmSetRedAlarmVO) throws BizException {
        AlarmPO alarmPO = alarmDAO.selectAlarmByDeviceIdAndAlarmStatus(deviceIotVO.getDeviceId(), EAlarmStatus.COMPLETED.getKey(), dataNode, EAlarmDevType.DEV.getKey(), null);
//        AlarmSetVO alarmSetRedAlarmVO = alarmSetDAO.selectAlarmSetBySwitchAndEffectAndNameAndLevel(devPointValue, EAlarmLevel.ALARM_RED.getKey(), dataNode);
        if (Objects.nonNull(alarmPO)) {
            alarmPO.setDataNode(dataNode);

            if (EAlarmLevel.ALARM_YELLOW.getKey() == alarmPO.getAlarmLevel()
                    && (1 == val.compareTo(new BigDecimal(alarmSetRedAlarmVO.getPhMax()))
                    || !(1 == val.compareTo(new BigDecimal(alarmSetRedAlarmVO.getPhMin()))))) {
                alarmPO.setAlarmLevel(EAlarmLevel.ALARM_RED.getKey());
                alarmPO.setAlarmInfo(String.format(devPointValue + QUALITY_UN, val));
                alarmDAO.updateAlarmInfoAndLevelAndEndTimeById(alarmPO);
                //处理系统消息
                handleAlarmSysNews(Constant.RISE_ALARM_HANDLE_SYS_IDEA, alarmPO.getId(), EAlarmHandleType.SYSTEM_HANDLE.getKey(), dataNode);
                //是否降级黄色
            } else if (EAlarmLevel.ALARM_RED.getKey() == alarmPO.getAlarmLevel()
                    && (!(1 == val.compareTo(new BigDecimal(alarmSetRedAlarmVO.getPhMax())))
                    || (1 == val.compareTo(new BigDecimal(alarmSetRedAlarmVO.getPhMin()))))) {
                alarmPO.setAlarmLevel(EAlarmLevel.ALARM_YELLOW.getKey());
                alarmPO.setAlarmInfo(String.format(devPointValue + QUALITY_UN, val));
                alarmDAO.updateAlarmInfoAndLevelAndEndTimeById(alarmPO);
                //处理系统消息
                handleAlarmSysNews(Constant.DROP_ALARM_HANDLE_SYS_IDEA, alarmPO.getId(), EAlarmHandleType.SYSTEM_HANDLE.getKey(), dataNode);
            }
            //处理曲线快照
            qualityAlarmAddAlarmCurveSnap(now, val, alarmPO.getId(), dataNode);
        }
        if (!Objects.nonNull(alarmPO)) {
            alarmPO = new AlarmPO();
            alarmPO.setDataNode(dataNode);
            alarmPO.setStationId(deviceIotVO.getStationId());
            alarmPO.setStationName(deviceIotVO.getStationName());
            alarmPO.setAlarmStatus(EAlarmStatus.UNREAD.getKey());
            alarmPO.setAlarmType(deviceIotVO.getAlarmTypeId());
            alarmPO.setAlarmTypeName(deviceIotVO.getAlarmTypeName());
            alarmPO.setAlarmInfo(String.format(devPointValue + QUALITY_UN, val));
            alarmPO.setDevType(EAlarmDevType.DEV.getKey());
            alarmPO.setGatherDevPoiId(deviceIotVO.getGatherDevPointId());

            if (1 == val.compareTo(new BigDecimal(alarmSetRedAlarmVO.getPhMax()))
                    || !(1 == val.compareTo(new BigDecimal(alarmSetRedAlarmVO.getPhMin())))) {
                alarmPO.setAlarmLevel(EAlarmLevel.ALARM_RED.getKey());
            } else {
                AlarmSetVO alarmSetYeAlarmVO = alarmSetDAO.selectAlarmSetBySwitchAndEffectAndNameAndLevel(devPointValue, EAlarmLevel.ALARM_YELLOW.getKey(), dataNode);
                if (1 == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getPhMax()))
                        || !(1 == val.compareTo(new BigDecimal(alarmSetYeAlarmVO.getPhMin()))))
                    alarmPO.setAlarmLevel(EAlarmLevel.ALARM_YELLOW.getKey());
            }
            alarmPO.setStartTime(now);
            alarmPO.setDeviceId(deviceIotVO.getDeviceId());
            alarmPO.setDataNode(dataNode);
            if (null != alarmPO.getAlarmLevel()) {
                alarmPO.setAlarmNum(String.format("SZ" + LocalDateTimeUtils.localDateToString(new Date()) + "%03d", sequenceDefineServiceImpl.increment(Constant.ALARM_NUM_SN_KEY, dataNode)));
                alarmPO.setDutyUserNames(deviceIotVO.getDutyUserNames());
                Long aId = alarmDAO.selsectNextAlarmId(dataNode);
                alarmPO.setId(aId);

                alarmDAO.insert(alarmPO);
                //处理告警曲线快照
                qualityAlarmAddAlarmCurveSnap(now, val, alarmPO.getId(), dataNode);
                //处理系统消息
                handleAlarmSysNews(EAlarmLevel.ALARM_YELLOW.getKey() == alarmPO.getAlarmLevel() ? Constant.YE_ALARM_HANDLE_SYS_IDEA : Constant.RED_ALARM_HANDLE_SYS_IDEA,
                        alarmPO.getId(), EAlarmHandleType.SYSTEM_HANDLE.getKey(), dataNode);

                //push message
                try {
                    List<Long> userIds = messageSubscribeDAO.selectMessageByStationIdAndType(deviceIotVO.getStationId(), MessageSubscribeType.ALARM_EVENT_PUSH.getKey(), dataNode);
                    if (null != userIds && userIds.size() > 0) {
                        //                    List<String> openId = userDAO.findOpenId(Arrays.asList((Long[]) ConvertUtils.convert(deviceIotVO.getOpeLeaderId().split(","), Long.class)), null);
                        List<String> openId = userDAO.findOpenId(userIds, dataNode);
                        AlarmPO finalAlarmPO = alarmPO;
                        openId.forEach(item -> {
                            try {
                                if (StringUtils.isNotEmpty(item))
                                    messageSubscribeServiceImpl.pushAlarm(item,
                                            aId,
                                            finalAlarmPO.getAlarmInfo(),
                                            "告警类型告警通知",
                                            "请尽快处理，点击查看详情",
                                            MessageSubscribeType.ALARM_EVENT_PUSH.getKey());

                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        });
                    }
                } catch (Exception e) {
                    logger.info("告警推送失败！", e);
                }
            }
        }
    }

    private void qualityAlarmNotPh(LocalDateTime now, DeviceIotVO deviceIotVO, BigDecimal val, String devPointValue, int seasonNumber, Double qualityAlarmParV, String dataNode) throws BizException {
        AlarmPO alarmPO = alarmDAO.selectAlarmByDeviceIdAndAlarmStatus(deviceIotVO.getDeviceId(), EAlarmStatus.COMPLETED.getKey(), dataNode, EAlarmDevType.DEV.getKey(), null);
        //告警设置
        if (Objects.nonNull(alarmPO)) {
            alarmPO.setDataNode(dataNode);
            //需要处理是否升级红色
            if (EAlarmLevel.ALARM_YELLOW.getKey() == alarmPO.getAlarmLevel() && 1 == val.compareTo(new BigDecimal(qualityAlarmParV))) {
                alarmPO.setAlarmLevel(EAlarmLevel.ALARM_RED.getKey());
                alarmPO.setAlarmInfo(String.format(devPointValue + QUALITY_UN, val));
                alarmDAO.updateAlarmInfoAndLevelAndEndTimeById(alarmPO);
                //处理系统消息
                handleAlarmSysNews(Constant.RISE_ALARM_HANDLE_SYS_IDEA, alarmPO.getId(), EAlarmHandleType.SYSTEM_HANDLE.getKey(), dataNode);
                //是否降级黄色
            } else if (EAlarmLevel.ALARM_RED.getKey() == alarmPO.getAlarmLevel() && !(1 == val.compareTo(new BigDecimal(qualityAlarmParV)))) {
                alarmPO.setAlarmLevel(EAlarmLevel.ALARM_YELLOW.getKey());
                alarmPO.setAlarmInfo(String.format(devPointValue + QUALITY_UN, val));
                alarmDAO.updateAlarmInfoAndLevelAndEndTimeById(alarmPO);
                //处理系统消息
                handleAlarmSysNews(Constant.DROP_ALARM_HANDLE_SYS_IDEA, alarmPO.getId(), EAlarmHandleType.SYSTEM_HANDLE.getKey(), dataNode);
            }
            //已有没处理完的告警，不管是否升级或降级告警，都要处理曲线快照
            qualityAlarmAddAlarmCurveSnap(now, val, alarmPO.getId(), dataNode);
        }

        if (Objects.isNull(alarmPO)) {
            alarmPO = new AlarmPO();
            alarmPO.setStationId(deviceIotVO.getStationId());
            alarmPO.setStationName(deviceIotVO.getStationName());
            alarmPO.setAlarmStatus(EAlarmStatus.UNREAD.getKey());
            alarmPO.setAlarmType(deviceIotVO.getAlarmTypeId());
            alarmPO.setAlarmTypeName(deviceIotVO.getAlarmTypeName());
            alarmPO.setAlarmInfo(String.format(devPointValue + QUALITY_UN, val));
            alarmPO.setDevType(EAlarmDevType.DEV.getKey());
            alarmPO.setGatherDevPoiId(deviceIotVO.getGatherDevPointId());
            if (1 == val.compareTo(new BigDecimal(qualityAlarmParV))) {
                alarmPO.setAlarmLevel(EAlarmLevel.ALARM_RED.getKey());
            } else {
                String devPointValueNoNum = null;
                if (devPointValue.contains(EQualityAlarmPar.NH3N.getKey())
                        || devPointValue.contains(EQualityAlarmPar.BOD5.getKey())) devPointValueNoNum = devPointValue;
                else devPointValueNoNum = StringUtil.removeStrMatcher(devPointValue);

                AlarmSetVO alarmSetYeAlarmVO = alarmSetDAO.selectAlarmSetBySwitchAndEffectAndNameAndLevel(devPointValueNoNum, EAlarmLevel.ALARM_YELLOW.getKey(), dataNode);

                if (EQualityAlarmPar.COD.getKey().equals(devPointValueNoNum))
                    qualityAlarmParV = alarmSetYeAlarmVO.getCod();
                if (EQualityAlarmPar.BOD5.getKey().equals(devPointValueNoNum))
                    qualityAlarmParV = alarmSetYeAlarmVO.getBod5();
                if (EQualityAlarmPar.SS.getKey().equals(devPointValueNoNum))
                    qualityAlarmParV = alarmSetYeAlarmVO.getSs();
                if (EQualityAlarmPar.TP.getKey().equals(devPointValueNoNum))
                    qualityAlarmParV = alarmSetYeAlarmVO.getTp();
                if (EQualityAlarmPar.TN.getKey().equals(devPointValueNoNum))
                    qualityAlarmParV = alarmSetYeAlarmVO.getTn();
                if (EQualityAlarmPar.NH3N.getKey().equals(devPointValueNoNum) && (seasonNumber >= 4 && seasonNumber <= 9))//北半球4-9 夏令时  10-3是冬令时
                    qualityAlarmParV = alarmSetYeAlarmVO.getNh3NS();
                if (EQualityAlarmPar.NH3N.getKey().equals(devPointValueNoNum) && (seasonNumber >= 10 || seasonNumber <= 3))
                    qualityAlarmParV = alarmSetYeAlarmVO.getNh3NW();

                if (!(1 == val.compareTo(new BigDecimal(qualityAlarmParV)))) {
                    logger.debug("{},小于或等于自定义值，不处理告警", deviceIotVO.getDeviceName());
                    return;
                }
                alarmPO.setAlarmLevel(EAlarmLevel.ALARM_YELLOW.getKey());
            }
            alarmPO.setStartTime(now);
            alarmPO.setDeviceId(deviceIotVO.getDeviceId());
            alarmPO.setDataNode(dataNode);
            if (null != alarmPO.getAlarmLevel()) {
                alarmPO.setAlarmNum(String.format("SZ" + LocalDateTimeUtils.localDateToString(new Date()) + "%03d", sequenceDefineServiceImpl.increment(Constant.ALARM_NUM_SN_KEY, dataNode)));
                alarmPO.setDutyUserNames(deviceIotVO.getDutyUserNames());

                Long aId = alarmDAO.selsectNextAlarmId(dataNode);
                alarmPO.setId(aId);
                alarmDAO.insert(alarmPO);
                //处理告警曲线快照
                qualityAlarmAddAlarmCurveSnap(now, val, alarmPO.getId(), dataNode);
                //处理系统消息
                handleAlarmSysNews(EAlarmLevel.ALARM_YELLOW.getKey().equals(alarmPO.getAlarmLevel()) ? Constant.YE_ALARM_HANDLE_SYS_IDEA : Constant.RED_ALARM_HANDLE_SYS_IDEA,
                        alarmPO.getId(), EAlarmHandleType.SYSTEM_HANDLE.getKey(), dataNode);

                //push message
                try {
                    List<Long> userIds = messageSubscribeDAO.selectMessageByStationIdAndType(deviceIotVO.getStationId(), MessageSubscribeType.ALARM_EVENT_PUSH.getKey(), dataNode);
                    if (null != userIds && userIds.size() > 0) {
                        //                    List<String> openId = userDAO.findOpenId(Arrays.asList((Long[]) ConvertUtils.convert(deviceIotVO.getOpeLeaderId().split(","), Long.class)), null);
                        List<String> openId = userDAO.findOpenId(userIds, dataNode);
                        AlarmPO finalAlarmPO = alarmPO;
                        openId.forEach(item -> {
                            try {
                                if (StringUtils.isNotEmpty(item))
                                    messageSubscribeServiceImpl.pushAlarm(item,
                                            aId,
                                            finalAlarmPO.getAlarmInfo(),
                                            "告警类型告警通知",
                                            "请尽快处理，点击查看详情",
                                            MessageSubscribeType.ALARM_EVENT_PUSH.getKey());

                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        });
                    }
                } catch (Exception e) {
                    logger.info("告警推送失败！", e);
                }
            }
        }
    }

    /**
     * 生成redis数据
     *
     * @param redisDataList
     * @param sensorValueMap
     * @param time
     * @param deviceIotVO
     */
    private void generateRedisDate(List<Map<String, Map<byte[], byte[]>>> redisDataList,
                                   Map<String, Object> sensorValueMap,
                                   String time,
                                   DeviceIotVO deviceIotVO,
                                   String netNum) {
        Map<byte[], byte[]> sensorTempMap = new HashMap<>();
        sensorTempMap.put("devId".getBytes(), deviceIotVO.getGatherDevPointId().toString().getBytes());
        sensorTempMap.put("collectTime".getBytes(), time.getBytes());
        String val = sensorValueMap.get("val").toString();
        sensorTempMap.put("val".getBytes(), normData(val).getBytes());

        // key=aaa:spa:物联网网关_规范
        Map<String, Map<byte[], byte[]>> dataMap = new HashMap<>();
        dataMap.put(DEVICE_KEY + netNum + "_" + sensorValueMap.get("key"), sensorTempMap);
        redisDataList.add(dataMap);
    }

    /**
     * 生成数据库存储
     *
     * @param pgDataMap
     * @param deviceIotVO
     * @param val
     */
    private void generatePgDate(Map<Long, Map<String, Object>> pgDataMap,
                                DeviceIotVO deviceIotVO,
                                String val) {
        Map<String, Object> realTimeData = Maps.newHashMap();
        Long devId = deviceIotVO.getGatherDevPointId();
        //处理成数据库录入的精度
        realTimeData.put(deviceIotVO.getGatherDevPointValue(), new BigDecimal(NumberUtil.precision(normData(val), deviceIotVO.getPrecision(), ZEROS)));
        pgDataMap.put(devId, realTimeData);
    }

    private void generatePgVal(List<GatherDevPointPO> pgValList,
                               DeviceIotVO deviceIotVO,
                               String val) {
        GatherDevPointPO gdpp = new GatherDevPointPO();
        gdpp.setId(deviceIotVO.getGatherDevPointId());
        gdpp.setCollectTime(LocalDateTime.now());
        //处理成数据库录入的精度
        gdpp.setParValue(new BigDecimal(NumberUtil.precision(normData(val), deviceIotVO.getPrecision(), ZEROS)).doubleValue());
        pgValList.add(gdpp);
    }

    /**
     * 处理报警数据
     * <p>
     * 对数据库操作的相关部分，会调到用getSQLPerfix对数据库前缀进行拼接
     * </p>
     *
     * @param map {"devId":"achanzhang666","data":[{"val":666,"key":"instantFlow","id":"ach1"},{"val":586,"key":"forwardCumulativeFlow","id":"ach2"},{"val":123,"key":"oppositeCumulativeFlow","id":"ach3"}],"time":"2020-01-07 14:52:00"}
     */
    @Transactional(rollbackFor = Exception.class)
    public void handleAlarmData(Map<String, Object> map) {
        long startTime = System.currentTimeMillis();
        try {
            String time = MapUtils.getString(map, "time");
            String netNum = MapUtils.getString(map, "devId", Constant.EMPTY);
            String tenantId = MapUtils.getString(map, "tenantId");
            LocalDateTime now = LocalDateTime.now();

            //tenantId为null则为一体机反之为云版本
            String dataNode = null;
            if (StringUtils.isNotEmpty(tenantId)) {
                Map<String, Object> mget = redisService.mget(REDIS_TENANT + tenantId, Object.class);
                if (null == mget) {
                    logger.info("未找到该租户id={}的dataNode,不处理告警", tenantId);
                    return;
                }
                Map<String, Object> dataNodeMap = (Map<String, Object>) JSON.parse((String) mget.get("dataNode"));
                dataNode = dataNodeMap.get(Constant.APPLICATION_NAME) + "";
                if (StringUtils.isBlank(dataNode) || dataNode.equals("null")) {
                    logger.info("该dataNode={}为空,不处理告警", dataNode);
                    return;
                }

                //校验数据是否在租户的站点下
                dataNode = getSQLPerfix(dataNode);
            }

            List<Map<String, Object>> dataList = (List<Map<String, Object>>) MapUtils.getObject(map, "data", Lists.newArrayList());
            if (dataList.isEmpty()) {
                logger.info("dataList={}为空,不处理数据", dataList);
                return;
            }
            // 组装解析后的值存入redis
            List<Map<String, Map<byte[], byte[]>>> redisDataList = new ArrayList<>();
            // 组装解析后的值存入数据库
            Map<Long, Map<String, Object>> pgDataMap = Maps.newHashMap();
            // 组装点位最新的值存入数据库
            List<GatherDevPointPO> pgValList = Lists.newArrayList();
            // 组装入库设备信息
            List<DevStoragePO> pgDsList = Lists.newArrayList();
            // 组装碳源累计采集
            AtomicLong carbonSourceAddGatherNum = new AtomicLong(0L);
            AtomicLong exceptPhosphorusAddGatherNum = new AtomicLong(0L);
            AtomicLong explosiveGasAddGatherNum = new AtomicLong(0L);
            List<String> tyPoints = aiModelParamDAO.findParByModeName("智能碳源");
            List<String> clPoints = aiModelParamDAO.findParByModeName("智能除磷");
            List<String> bqPoints = aiModelParamDAO.findParByModeName("智能曝气");

            //遍历传输过来的数据，解析处理
            for (Map<String, Object> dataMap : dataList) {
//            for (int i = 0, len = dataList.size(); i < len; i++) {
//                Map<String, Object> dataMap = dataList.get(i);
                if (MapUtils.isEmpty(dataMap)) {
                    logger.info("设备data里没有数据,不处理告警");
                    continue;
                }

                String key = MapUtils.getString(dataMap, "key");
                String vaStr = MapUtils.getString(dataMap, "val");

                if (Objects.isNull(vaStr)) {
                    logger.info("{}设备data的val为空,不处理告警", vaStr);
                    continue;
                }

                logger.debug("{}接收物联网点位key，方便排查点位规范正确否：", key);
                String[] keys = key.split("_");
                if (keys.length < 2 || keys.length < 3) continue;
                String d_change = D_CHANGE.get(keys[2]);
                String newKey = null;
                if (!StringUtils.isEmpty(d_change)) {
                    newKey = keys[0] + "_" + keys[1] + "_" + d_change + "_" + keys[3];
                    dataMap.put("key", newKey);
                    //替换值
                    if (d_change.equals(K) || d_change.equals(ZD) || d_change.equals(YC) || d_change.equals(QD) || d_change.equals(GZ) || d_change.equals(FK))
                        vaStr = ONE;
                    if (d_change.equals(G) || d_change.equals(SD) || d_change.equals(BD) || d_change.equals(TZ) || d_change.equals(ZC) || d_change.equals(FG))
                        vaStr = ZEROS;
                } else newKey = key;

                //校验设备
//                DeviceIotVO deviceIotVO = deviceServiceImpl.selectDeviceByIotGatewayNumAndKey(netNum, newKey, dataNode);
                DeviceIotVO deviceIotVO = redisService.getJson(MQ_DEVICE_KEY + netNum + "-" + newKey, DeviceIotVO.class);
                if (Objects.isNull(deviceIotVO)) {
                    logger.info("{}的设备对象为空,可能未领用设备,不处理告警", deviceIotVO);
                    continue;
                }

                // 生成redis数据（取分割后第一个来缓存）
                generateRedisDate(redisDataList, dataMap, time, deviceIotVO, netNum);
                // 生成数据库实时数据
                generatePgDate(pgDataMap, deviceIotVO, vaStr);
                //生成控制类设备参数数据
                generatePgVal(pgValList, deviceIotVO, vaStr);
                //生成入库设备初始工作时间数据
                generateDsVal(pgDsList, deviceIotVO, vaStr);
                //碳源累计统计
                generateCarbonSourceAddGatherNum(carbonSourceAddGatherNum, exceptPhosphorusAddGatherNum, explosiveGasAddGatherNum, deviceIotVO, tyPoints, clPoints, bqPoints);
                if (EPlcLogicPointType.Y_PLC_LOGIC_POINT.getKey().equals(deviceIotVO.getPlcLogicPoint()) && !keys[0].equals(CSQ) && !keys[0].equals(JSK)) {
                    logger.info("{}为逻辑点位,不处理告警及关闭告警", newKey);
                    continue;
                }
                //在循环里面起线程异步检测是否自动关闭告警（如果在循环外面处理会遇到双层循环性能更低)
                sysCloseAlarmThread(dataNode, vaStr, deviceIotVO, deviceIotVO.getGatherDevPointValue());
                // 生成告警记录（处理告警信息、处理告警快照）
                generateAlarm(dataNode, now, vaStr, deviceIotVO, deviceIotVO.getGatherDevPointValue());
            }
            redisCarbonSourceAddGatherNumData(carbonSourceAddGatherNum.get(), exceptPhosphorusAddGatherNum.get(), explosiveGasAddGatherNum.get());
            redisBatchData(redisDataList);
            deviceTimeDataServiceImpl.handlePgData(pgDataMap, time, dataNode, now);
            gatherDevServiceImpl.handlePgVal(pgValList, dataNode);
            devStorageServiceImpl.handlePgVal(pgDsList, dataNode);
            pointCalculationRuleService.calculate(tenantId, dataNode, time, dataList);
            long endTime = System.currentTimeMillis();
            System.out.println("mq处理数据的时长==========================================================：" + (endTime - startTime));
        } catch (Exception e) {
            logger.error("物联网mq数据处理失败！！！", e);
        }
    }

    private void generateCarbonSourceAddGatherNum(AtomicLong carbonSourceAddGatherNum, AtomicLong exceptPhosphorusAddGatherNum, AtomicLong explosiveGasAddGatherNum,
                                                  DeviceIotVO deviceIotVO, List<String> tyPoints, List<String> clPoints, List<String> bqPoints) {
        if (null != tyPoints && tyPoints.size() > 0)
            if (tyPoints.contains(deviceIotVO.getGatherDevPointValue()))
                carbonSourceAddGatherNum.set(carbonSourceAddGatherNum.get() + 1);

        if (null != clPoints && clPoints.size() > 0)
            if (clPoints.contains(deviceIotVO.getGatherDevPointValue()))
                exceptPhosphorusAddGatherNum.set(exceptPhosphorusAddGatherNum.get() + 1);

        if (null != bqPoints && bqPoints.size() > 0)
            if (bqPoints.contains(deviceIotVO.getGatherDevPointValue()))
                explosiveGasAddGatherNum.set(explosiveGasAddGatherNum.get() + 1);
    }

    @Async("asyncServiceExecutor")
    public void redisCarbonSourceAddGatherNumData(Long carbonSourceAddGatherNum, Long exceptPhosphorusAddGatherNum, Long explosiveGasAddGatherNum) {
        if (null != carbonSourceAddGatherNum && carbonSourceAddGatherNum > 0)
            redisService.increment(CARBON_SOURCE_ADD_GATHER_NUM, carbonSourceAddGatherNum);
        if (null != exceptPhosphorusAddGatherNum && exceptPhosphorusAddGatherNum > 0)
            redisService.increment(EXCEPT_PHOSPHORUS_ADD_GATHER_NUM, exceptPhosphorusAddGatherNum);
        if (null != explosiveGasAddGatherNum && explosiveGasAddGatherNum > 0)
            redisService.increment(EXPLOSIVE_GAS_ADD_GATHER_NUM, explosiveGasAddGatherNum);
    }

    @Async("asyncServiceExecutor")
    public void redisBatchData(List<Map<String, Map<byte[], byte[]>>> redisDataList) {
        redisService.batchData(redisDataList, expire);
    }

    private void handleAlarmSysNews(String handleIdea, Long alarmId, Integer handleType, String dataNode) {
        alarmHandleServiceImpl.handleAlarmSysNews(handleIdea, alarmId, handleType, null, null, dataNode);
    }

    /**
     * 规范数据 排除 true和false的情况 规范成1和0
     *
     * @param val
     * @return
     */
    private String normData(String val) {
        if (val.equalsIgnoreCase(Constant.TRUE_STR)) val = "1";
        if (val.equalsIgnoreCase(FALSE_STR)) val = "0";
        return val;
    }

    private void generateDsVal(List<DevStoragePO> pgValList,
                               DeviceIotVO deviceIotVO,
                               String val) {
        String point = deviceIotVO.getGatherDevPointValue().split("_")[2];
        if (LJYXSJ.equals(point) && 0 == deviceIotVO.getInitWorkTime()) {
            DevStoragePO gdpp = new DevStoragePO();
            gdpp.setId(deviceIotVO.getSdsId());
            gdpp.setInitWorkTime(1);

            if (!Pattern.compile("[0-9]*").matcher(val).matches()) {
                logger.error("本条累计时间计算为小数不处理：", deviceIotVO.getGatherDevPointValue() + "值：" + val);
                return;
            }
            Integer cycleHour = deviceIotVO.getMaintenanceCycleHour();
            if (cycleHour == 0) return;

            Integer valI = Integer.valueOf(val);
            gdpp.setWorkTime(valI - (valI / cycleHour) * cycleHour);
            pgValList.add(gdpp);
        }
    }
}
